#ifndef _XDG_THREAD_POOL_BLOCKING_HPP_
#define _XDG_THREAD_POOL_BLOCKING_HPP_

/*
 * ThreadPool - 基于 moodycamel::BlockingConcurrentQueue 的线程池实现
 *
 * 要求:
 * - 最低 C++11
 * - 使用 moodycamel::BlockingConcurrentQueue<Task> 作为底层队列
 * - 有界/无界队列支持（m_queueSize == -1 表示无界）
 * - submitTask 返回 std::future（使用 std::packaged_task）
 * - submit 不返回 future（在任务内部捕获异常并记录）
 * - 优雅关闭与立即关闭支持
 *
 * 设计要点（简述）:
 * - BlockingConcurrentQueue 提供阻塞出队，避免 busy-wait。
 * - 生产者负责在队列达到上限时进行阻塞（通过 condition_variable），以实现有界队列。
 * - 使用原子计数 m_queueCount 来跟踪队列中任务的精确数量（比 size_approx 更可靠）。
 * - worker 使用带超时的阻塞出队（wait_dequeue_timed）来定期检查停止标志，以便在 destroy 请求到达时能及时退出。
 *
 * 线程安全: 所有公开接口线程安全。
 *
 * 备注:
 * - 该实现避免裸指针、手动 new/delete；使用 shared_ptr 封装 packaged_task，以便安全传递给队列中的 lambda。
 * - 如果你完全信任 BlockingConcurrentQueue 的“可以被 close/unblock”特性并且想更低延迟（无需超时轮询），可以改用 wait_dequeue() + queue 的 close/unlock 特性（若库版本支持）。
 */

#include <vector>
#include <thread>
#include <mutex>
#include <atomic>
#include <condition_variable>
#include <functional>
#include <future>
#include <memory>
#include <stdexcept>
#include <chrono>
#include <iostream>

#include "./concurrentqueue/blockingconcurrentqueue.h"

namespace xdg {

#if __cplusplus >= 201703L
    template<typename F, typename... Args>
    using result_type = typename std::invoke_result<F, Args...>::type;
#else
    template<typename F, typename... Args>
    using result_type = typename std::result_of<F(Args...)>::type;
#endif

class ThreadPool {
public:
    /**
     * @param threadCount 线程数: 0 表示使用硬件并发 * 2
     * @param queueSize 队列容量: -1 表示无界，0 表示使用硬件并发 * 10
     *
     * 注意: 构造函数不会启动线程，需显式调用 start()
     */
    explicit ThreadPool(unsigned int threadCount = 0, int queueSize = -1);

    /**
     * 析构函数：尝试优雅关闭线程池（destroy(false)）
     * 析构函数绝对不会抛异常
     */
    ~ThreadPool();

    /**
     * 启动线程池（幂等）
     * 返回实际启动的线程数
     * 可能抛 std::runtime_error（线程创建失败）
     */
    int start();

    /**
     * 停止线程池
     * @param immediate true 表示立即停止（丢弃队列内未执行任务）
     *                  false 表示优雅关闭（等待队列中已有任务执行完毕）
     */
    void destroy(bool immediate = false);

    /**
     * 提交任务并返回 future
     * 如果线程池已停止，则抛出 std::runtime_error
     */
    template<typename F, typename... Args>
    auto submitTask(F&& f, Args&&... args) -> std::future< result_type<F, Args...> >;

    /**
     * 提交任务（不返回 future）
     * 如果线程池已停止，则抛出 std::runtime_error
     */
    template<typename F, typename... Args>
    void submit(F&& f, Args&&... args);

    unsigned int getThreadCount() const { return m_threadCount; }
    int getQueueSize() const { return m_queueSize; }
    int getStatus() const { return m_status.load(); } // 0=运行 -1=未启动或已停止

    // 禁止拷贝/移动
    ThreadPool(const ThreadPool&) = delete;
    ThreadPool& operator=(const ThreadPool&) = delete;
    ThreadPool(ThreadPool&&) = delete;
    ThreadPool& operator=(ThreadPool&&) = delete;

private:
    using Task = std::function<void()>;

    // 每个 worker 的主循环
    void workerThread();

private:
    // 配置参数
    unsigned int m_threadCount;
    int m_queueSize; // -1 表示无界

    // 状态标志
    std::atomic<int> m_status; // 0 = running, -1 = stopped/not started
    std::atomic<bool> m_stop;  // 停止标志

    // 底层队列（阻塞式队列）
    moodycamel::BlockingConcurrentQueue<Task> m_tasksQueue;

    // 工作线程容器
    std::vector<std::thread> m_workers;

    // 生产者阻塞控制（用于有界队列）
    std::mutex m_prod_mutex;
    std::condition_variable m_prod_cv;

    // 队列中任务的精确计数（原子），比 size_approx 更可靠
    std::atomic<size_t> m_queueCount;
};

/* ---------------- 实现部分 ---------------- */

inline ThreadPool::ThreadPool(unsigned int threadCount, int queueSize)
    : m_threadCount(0),
      m_queueSize(queueSize),
      m_status(-1),
      m_stop(true), // 初始化为已停止（尚未 start）
      m_tasksQueue(),
      m_workers(),
      m_queueCount(0)
{
    unsigned int cores = std::thread::hardware_concurrency();
    if (cores == 0) cores = 2; // 兜底

    m_threadCount = (threadCount == 0) ? cores * 2 : threadCount;

    if (queueSize == 0) {
        m_queueSize = static_cast<int>(cores * 10);
    }

    // m_status 默认为 -1（未启动）
}

/*
 * 析构：优雅关闭
 * - 在析构中调用 destroy(false)
 * - 捕获并忽略任何异常，保证析构函数不抛出
 */
inline ThreadPool::~ThreadPool() {
    try {
        destroy(false);
    } catch (...) {
        // 不抛出
    }
}

/*
 * start:
 * - 幂等：如果已经正在运行，直接返回线程数
 * - 创建线程并设置状态
 * - 如果线程创建失败，回滚并抛出异常
 */
inline int ThreadPool::start() {
    // 如果已经运行，直接返回
    if (m_status.load() == 0) {
        return static_cast<int>(m_threadCount);
    }

    // 准备状态
    m_stop.store(false);
    m_queueCount.store(0);

    // 清理旧线程（保险）
    for (auto &t : m_workers) {
        if (t.joinable()) t.join();
    }
    m_workers.clear();
    m_workers.reserve(m_threadCount);

    try {
        for (unsigned int i = 0; i < m_threadCount; ++i) {
            m_workers.emplace_back(&ThreadPool::workerThread, this);
        }
        m_status.store(0); // 运行中
        return static_cast<int>(m_threadCount);
    } catch (...) {
        // 回滚：设置停止，join 已创建的线程
        m_stop.store(true);
        for (auto &t : m_workers) {
            if (t.joinable()) t.join();
        }
        m_workers.clear();
        m_status.store(-1);
        throw std::runtime_error("ThreadPool start: failed to create threads");
    }
}

/*
 * destroy:
 * - immediate == false: 优雅关闭（等待队列中已有任务执行完成）
 * - immediate == true : 立即停止（丢弃队列中未执行任务）
 *
 * 注意：
 * - 因为 BlockingConcurrentQueue::wait_dequeue 可能会一直阻塞，
 *   我们在 worker 中使用带超时的 wait_dequeue_timed 来定期检查 m_stop，因此这里不需要显式唤醒队列。
 */
inline void ThreadPool::destroy(bool immediate) {
    // 如果未启动或已停止，直接返回（幂等）
    if (m_status.load() != 0) return;

    // 1) 设置停止标志
    m_stop.store(true);

    if (immediate) {
        // 2) 丢弃队列中的所有任务（出队直到空），同时维护计数
        Task tmp;
        while (m_tasksQueue.try_dequeue(tmp)) {
            // 成功出队则减少计数
            size_t prev = m_queueCount.fetch_sub(1);
            (void)prev; // 仅为计数维护，避免编译器警告
            // Task 的析构会释放资源
        }
    } else {
        // 优雅：不丢弃，等待 worker 处理完即可
        // m_stop 为 true，会告知 worker 在处理完任务后退出
    }

    // 3) 唤醒可能在生产端等待队列满的线程
    m_prod_cv.notify_all();

    // 4) 等待 worker 退出
    for (auto &t : m_workers) {
        if (t.joinable()) t.join();
    }
    m_workers.clear();

    // 5) 标记为停止
    m_status.store(-1);
}

/*
 * workerThread:
 * - 使用 wait_dequeue_timed 来阻塞式等待新任务（带超时轮询，用于检查停止标志）
 * - 当拿到任务时执行（在任务执行完成后递减计数并 notify 一名等待的生产者）
 *
 * 设计说明：
 * - 我选用了短超时时间（100ms），在 stop 时可以较快响应退出。
 * - 也可根据需求把超时调整为更小或更大，或改用库提供的“关闭队列并唤醒所有 waiting consumers”的 API（若库支持）。
 */
inline void ThreadPool::workerThread() {
    const std::chrono::milliseconds waitTimeout(10); // 超时轮询间隔（可调）

    while (true) {
        // 1) 优先检查停止条件：如果 stop 且队列为空，则退出
        if (m_stop.load() && m_queueCount.load() == 0) {
            return;
        }

        // 2) 阻塞式出队（带超时）
        Task task;
        bool got = false;
        try {
            // wait_dequeue_timed 返回 true 表示成功出队，false 表示超时（未出队）
            got = m_tasksQueue.wait_dequeue_timed(task, waitTimeout);
        } catch (...) {
            // 有些实现可能抛异常（极少见），记录并继续或退出
            std::cerr << "[ThreadPool] Warning: exception from wait_dequeue_timed\n";
            got = false;
        }

        if (!got) {
            // 超时或出错：循环回到开头检查停止条件
            continue;
        }

        // 成功取出任务：执行任务（在执行前无需再修改队列计数）
        try {
            task();
        } catch (const std::exception &e) {
            std::cerr << "[ThreadPool] Exception in task: " << e.what() << std::endl;
        } catch (...) {
            std::cerr << "[ThreadPool] Unknown exception in task" << std::endl;
        }

        // 执行完成后，递减队列计数并通知一个生产者（若有等待）
        m_queueCount.fetch_sub(1);
        // 唤醒生产者（如果队列是有界且有线程阻塞在 submit）
        m_prod_cv.notify_one();
    }
}

/*
 * submitTask:
 * - 使用 std::packaged_task 将可调用对象包装起来，以获得 future
 * - 有界队列时：在入队前阻塞（通过 condition_variable）直到有空位或线程池停止
 * - 无界队列时：直接入队
 * - 如果线程池已停止，则抛出 std::runtime_error
 *
 * 内存安全：
 * - packaged_task 存储在 shared_ptr 中并传递给队列（lambda 捕获 shared_ptr），
 *   从而避免任务对象在队列中被移动/拷贝时造成悬空引用问题。
 */
template<typename F, typename... Args>
inline auto ThreadPool::submitTask(F&& f, Args&&... args) -> std::future< result_type<F, Args...> >
{
    if (m_stop.load()) {
        throw std::runtime_error("ThreadPool is stopped; cannot submit");
    }

    using ReturnType = result_type<F, Args...>;

    // 使用 std::packaged_task 封装任务并获取 future
    auto boundTask = std::bind(std::forward<F>(f), std::forward<Args>(args)...);
    auto pack = std::make_shared<std::packaged_task<ReturnType()>>(std::move(boundTask));
    std::future<ReturnType> fut = pack->get_future();

    // 将 shared_ptr 的 packaged_task 包装为 Task（可放入队列）
    Task wrapper = [pack]() {
        // 注意：packaged_task 会将异常保存到 future 中
        (*pack)();
    };

    // --- 入队（处理有界 / 无界） ---
    if (m_queueSize == -1) {
        // 无界：直接入队（BlockingConcurrentQueue 的 enqueue 通常不会失败）
        m_tasksQueue.enqueue(wrapper);
        m_queueCount.fetch_add(1);
        return fut;
    } else {
        // 有界：需要阻塞等待空位
        std::unique_lock<std::mutex> lk(m_prod_mutex);
        // 等待条件：队列未满 或 线程池停止
        m_prod_cv.wait(lk, [this] {
            return m_stop.load() || static_cast<int>(m_queueCount.load()) < m_queueSize;
        });

        if (m_stop.load()) {
            throw std::runtime_error("ThreadPool is stopped; cannot submit");
        }

        // 此时有空位：入队
        m_tasksQueue.enqueue(wrapper);
        m_queueCount.fetch_add(1);

        // unlock happens automatically
        return fut;
    }
}

/*
 * submit (不返回 future)
 * - 与 submitTask 类似，但直接包装调用并在任务内部捕获异常（不传递）
 */
template<typename F, typename... Args>
inline void ThreadPool::submit(F&& f, Args&&... args) {
    if (m_stop.load()) {
        throw std::runtime_error("ThreadPool is stopped; cannot submit");
    }

    // 将函数绑定成可无参调用对象
    auto bound = std::bind(std::forward<F>(f), std::forward<Args>(args)...);
    Task wrapper = [bound]() mutable {
        try {
            bound();
        } catch (const std::exception &e) {
            std::cerr << "[ThreadPool] Exception in submit task: " << e.what() << std::endl;
        } catch (...) {
            std::cerr << "[ThreadPool] Unknown exception in submit task" << std::endl;
        }
    };

    if (m_queueSize == -1) {
        // 无界：直接入队
        m_tasksQueue.enqueue(wrapper);
        m_queueCount.fetch_add(1);
        return;
    } else {
        // 有界：等待空位
        std::unique_lock<std::mutex> lk(m_prod_mutex);
        m_prod_cv.wait(lk, [this] {
            return m_stop.load() || static_cast<int>(m_queueCount.load()) < m_queueSize;
        });

        if (m_stop.load()) {
            throw std::runtime_error("ThreadPool is stopped; cannot submit");
        }

        m_tasksQueue.enqueue(wrapper);
        m_queueCount.fetch_add(1);
        return;
    }
}

} // namespace xdg

#endif // _XDG_THREAD_POOL_BLOCKING_HPP_
